package com.atguigu.edu.realtime.common.util;

import com.atguigu.edu.realtime.common.constant.Constant;

/**
 * @author WEIYUNHUI
 * @date 2024/12/23 10:41
 */
public class FlinkSqlUtil {

    public static  String getDorisSinkDDL( String database , String table ){
        return  " WITH ( " +
                "      'connector' = 'doris', " +
                "      'fenodes' = '" + Constant.DORIS_FENODES+  "'," +
                "      'table.identifier' = '" + database + "." + table +"' ," +
                "      'username' = '" + Constant.DORIS_USERNAME + "'," +
                "      'password' = '" + Constant.DORIS_PASSWORD+ "' ,  " +
                "      'sink.enable-2pc' = 'false' " +
        //"      'sink.label-prefix' = 'doris_label" + System.currentTimeMillis()+ "' " +
                " )" ;
    }

    public static String getUpsertKafkaSinkDDL( String topic ){
        return " WITH (" +
                "  'connector' = 'upsert-kafka'," +
                "  'topic' = '" + topic +  "'," +
                "  'properties.bootstrap.servers' = '"+ Constant.KAFKA_BROKERS  +"'," +
                "  'key.format' = 'json', " +
                "  'value.format' = 'json' " +
                ")" ;
    }

    public static String getKafkaSinkDDL( String topic ){
        return " WITH (" +
                "  'connector' = 'kafka'," +
                "  'topic' = '" + topic +  "'," +
                "  'properties.bootstrap.servers' = '"+ Constant.KAFKA_BROKERS  +"'," +
                "  'format' = 'json' ," +
                "  'sink.delivery-guarantee' = 'at-least-once' " +
                //"  'sink.delivery-guarantee' = 'exactly-once' , " +
                //"  'sink.transactional-id-prefix' = 'gmall-realtime-" + System.currentTimeMillis()+ "' , " +
                //  检查点超时时间 <= 生产者事务超时时间 <=  KafkaBroker最大事务超时时间
                //"  'transaction.timeout.ms' = '600000' " +
                ")" ;
    }

    public static String getKafkaSourceDDL( String topic , String groupId ){
        return "WITH (" +
                "  'connector' = 'kafka'," +
                "  'topic' = '" + topic+ "',\n" +
                "  'properties.bootstrap.servers' = '"+ Constant.KAFKA_BROKERS  +"'," +
                "  'properties.group.id' = '" + groupId + "'," +
                "  'scan.startup.mode' = 'earliest-offset'," +
                "  'format' = 'json' , " +
                "  'json.ignore-parse-errors' = 'true'" + // 当 json 解析失败的时候,忽略这条数据
                //"  'isolation.level' = 'read_committed'" +
                ")" ;
    }
}
